package com.jie.flink.cdc.doman;

import com.jie.flink.cdc.flinksink.config.ElasticsearchConfigProperties;
import com.jie.flink.cdc.flinksink.config.KafkaConfigProperties;
import com.jie.flink.cdc.flinksink.config.PulsarConfigProperties;
import com.jie.flink.cdc.flinksink.config.SinkConfigProperties;
import com.jie.flink.cdc.flinksink.config.SinkLogConfigProperties;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;

/**
 * @author zhanggj
 * @date 2023/5/25 14:09
 * @desc
 */
@Getter
public enum FlinkCdcSinkType {
    ES("es", ElasticsearchConfigProperties.class),
    KAFKA("kafka", KafkaConfigProperties.class),
    PULSAR("pulsar", PulsarConfigProperties.class),
    LOG("log", SinkLogConfigProperties.class),

    ;

    private String code;
    private Class<? extends SinkConfigProperties> sinkConfigClass;

    FlinkCdcSinkType(String code, final Class<? extends SinkConfigProperties> sinkConfigClass) {
        this.code = code;
        this.sinkConfigClass = sinkConfigClass;
    }

    public static FlinkCdcSinkType getByCode(String code) {
        if (StringUtils.isBlank(code)) {
            return null;
        }
        for (FlinkCdcSinkType type : FlinkCdcSinkType.values()) {
            if (type.code.equals(code)) {
                return type;
            }
        }
        return null;
    }

}
